package com.ibm.cps.join;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.RawDataMessage;
import com.ibm.cps.newmessage.MetadataInputSchema;
import com.ibm.cps.newmessage.MetadataInputSchema.FieldType;
import com.ibm.datamodel.timeseries.parser.JsonParser;
import com.ibm.interfaces.IMessageParser;
import com.ibm.interfaces.INoSqlDB;
import com.ibm.mongo.CollectionAccess;
import com.ibm.mongo.MongoNoSqlDB;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;
import com.mongodb.MongoClientURI;

public class MongoJoiner extends AbstractJoiner implements KryoSerializable{

	/*
	 * this url should provide collection's name.
	 */
	public String mongo_static_url;
	public String dsName;
	
	public INoSqlDB mongodb;
	private CollectionAccess colAccess;
	private int index;
	private String keyName;

	
	public MongoJoiner(String metadata) throws CPSException{
		super(metadata);
		try {
			this.dsName = JSONUtility.getString(MessageFields.NAME, metadata);
			ObjectNode node = JSONUtility.fromObjectString(metadata);
			mongo_static_url = JSONUtility.GetString(MessageFields.STATIC_MONGO_URL, node);
			MongoClientURI url = new MongoClientURI(mongo_static_url);
			this.mongodb = new MongoNoSqlDB(mongo_static_url, url.getDatabase());
			colAccess = new CollectionAccess(mongodb);
			index = mongodb.getCollection(url.getCollection());
			keyName = node.get(MessageFields.KeyName).asText();
		} catch (CPSException e) {
			e.printStackTrace();
		}
		
	}



	@Override
	public void close() {
		try {
			this.mongodb.close();
		} catch (CPSException e) {
			e.printStackTrace();
		}
	}

	@Override
	public Collection<ParsedDataMessage> query(ObjectNode node) {
		Collection<ParsedDataMessage> collection = new ArrayList<>();
		try {
			Collection<String> results = colAccess.get(index, node, keyName);
			if (results.size() == 0) {
				return collection;
			}
			Long tsKey = node.get(MessageFields.TSKEY).asLong();
			for (String result : results) {
				ObjectNode resultNode = JSONUtility.fromObjectString(result);
				String fullQualifyresult = fullQualifyresult(resultNode);
				RawDataMessage raw = new RawDataMessage(dsmetadata.getTenantid(), dsmetadata.getSchemaid(), 
						dsmetadata.getName(), tsKey, fullQualifyresult.getBytes());
				IMessageParser parser = new JsonParser(schema);
				ParsedDataMessage message = parser.parseDataMessage(raw);
				collection.add(message);
			}
			
		} catch (CPSException e) {
			e.printStackTrace();
		}
		return collection;
	}
	
	private String fullQualifyresult(ObjectNode resultNode){
		ObjectNode fullQualifyNode = JSONUtility.newObjectNode();
		Iterator<Entry<String, JsonNode>> iterator = resultNode.fields();
		while (iterator.hasNext()) {
			Map.Entry<String, JsonNode> entry = iterator.next();
			fullQualifyNode.set(dsName + "." + entry.getKey(), entry.getValue());
		}
		return fullQualifyNode.toString();
	}


	@Override
	public void read(Kryo arg0, Input arg1) {
		mongo_static_url = arg1.readString();
		dsName = arg1.readString();
		keyName = arg1.readString();
		MongoClientURI url = new MongoClientURI(mongo_static_url);
		mongodb = new MongoNoSqlDB(mongo_static_url, url.getDatabase());
		colAccess = new CollectionAccess(mongodb);
		try {
			index = mongodb.getCollection(url.getCollection());
		} catch (CPSException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void write(Kryo arg0, Output arg1) {
		arg1.writeString(mongo_static_url);
		arg1.writeString(dsName);
		arg1.writeString(keyName);
		try {
			mongodb.close();
		} catch (CPSException e) {
			e.printStackTrace();
		}
	}

	@Override
	public MetadataInputSchema getInputSchema() {
		MetadataInputSchema schema = new MetadataInputSchema();
		schema.addField(FieldType.String, MessageFields.STATIC_MONGO_URL,MessageFields.NAME);
		return schema;
	}

}
